package com.whale.cloud.util.hbase;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by jiaoshiru on 2017/6/5.
 */
@Log4j2
public class SpringHbaseClient {

    private Configuration configuration;

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    private HbaseTemplate hbaseTemplate;

    public void setHbaseTemplate(HbaseTemplate hbaseTemplate) {
        this.hbaseTemplate = hbaseTemplate;
    }



    /**
     * 新建表
     * @param tableName 表名
     * @param family 列族名
     */
    public void creatTable(String tableName, String[] family){
        try{
            HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);

            HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
            for (int i = 0; i < family.length; ++i) {
                tableDescriptor.addFamily(new HColumnDescriptor(family[i]));
            }

            if (hBaseAdmin.tableExists(tableName)) {
                log.debug("Table exists");
            }else {
                hBaseAdmin.createTable(tableDescriptor);
                log.debug("Create table success");
            }
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 新增数据
     *
     * @param rowKey 主键
     * @param tableName 表名
     * @param familyName 列族名
     * @param column 列名
     * @param value 值
     * @return
     */
    public int add(String tableName, String rowKey, String familyName, String column, String value) {
        byte flag = 0;
        try {
            hbaseTemplate.put(tableName, rowKey, familyName, column, Bytes.toBytes(value));
            flag = 1;
            return flag;
        } catch (Exception e) {
            log.debug(e.getMessage());
            return flag;
        }
    }

    /**
     * 删除数据
     *
     * @param tableName 表名
     * @param rowKey 主键
     * @param falilyName 列族名
     * @param columnName 列名
     * @throws IOException
     */
    public void delete(String tableName, String rowKey, String falilyName, String columnName) throws IOException {
        hbaseTemplate.delete(tableName, rowKey, falilyName, columnName);
    }

    /**
     * 更新数据
     *
     * @param tableName 表名
     * @param rowKey 主键
     * @param familyName 列族名
     * @param columnName 列名
     * @param value 值
     * @return
     */
    public int update(String tableName, String rowKey, String familyName, String columnName, String value) {
        byte flag = 0;
        try {
            hbaseTemplate.put(tableName, rowKey, familyName, columnName, Bytes.toBytes(value));
            flag = 1;
            return flag;
        } catch (Exception e) {
            log.debug(e.getMessage());
            return flag;
        }
    }

    /**
     * 查询主键下所有列
     *
     * @param tableName 表名
     * @param rowKey 主键
     * @return
     * @throws IOException
     */
    public JSONArray getResult(String tableName, String rowKey) {
        JSONArray result = new JSONArray();
        result = hbaseTemplate.get(tableName, rowKey, new RowMapper<JSONArray>() {
            @Override
            public JSONArray mapRow(Result result, int rowNum) throws Exception {
                JSONArray results = new JSONArray();
                KeyValue[] keyValues = result.raw();
                int var = keyValues.length;

                for (int i = 0; i < var; ++i) {
                    KeyValue keyValue = keyValues[i];
                    String colum = new String(keyValue.getQualifier());
                    String values = new String(keyValue.getValue());
                    Long timestamp = new Long(keyValue.getTimestamp());
                    values = values.replace("NaN", "0");

                    try {
                        JSONObject e = JSONObject.parseObject(values);
                        Map m = (Map) JSONObject.parseObject(colum, HashMap.class);
                        e.putAll(m);
                        e.put("timestamp", timestamp);
                        results.add(e);
                    } catch (Exception e) {
                        ;
                    }
                }
                return results;
            }
        });
        return result;
    }

    /**
     * 查询指定列
     *
     * @param tableName 表名
     * @param rowKey 主键
     * @param falilyName 列族名
     * @param columnName 列名
     * @return
     * @throws IOException
     */

    public JSONObject getResultColumn(String tableName, String rowKey, String falilyName, String columnName) {
        JSONObject result = new JSONObject();
        hbaseTemplate.get(tableName, rowKey, falilyName, columnName, new RowMapper<JSONObject>() {
            @Override
            public JSONObject mapRow(Result result, int rowNum) throws Exception {
                JSONObject results = new JSONObject();
                List<Cell> ceList = result.listCells();
                if (ceList != null && ceList.size() > 0) {
                    String res = "";
                    if (result.size() > 0) {
                        byte[] f = new byte[]{" ".getBytes()[0]};
                        res = null == Bytes.toString(null == result.getValue(falilyName.getBytes(), columnName.getBytes()) ? f : result.getValue(falilyName.getBytes(), columnName.getBytes())) ? "" : Bytes.toString(result.getValue(falilyName.getBytes(), columnName.getBytes()));
                    }
                    if (res == null) {
                        res = "";
                    }

                    res = res.replace("NaN", "0");
                    results = JSONObject.parseObject(res.trim());
                }
                return results;
            }
        });
        return result;
    }

    /**
     * 查询多个列
     *
     * @param tableName 表名
     * @param rowKey 主键
     * @param columnName 多个列名
     * @return
     * @throws IOException
     */
    public JSONArray getResultColumnList(String tableName, String rowKey, List<String> columnName){
        JSONArray result = new JSONArray();
        result = hbaseTemplate.get(tableName, rowKey, new RowMapper<JSONArray>() {
            @Override
            public JSONArray mapRow(Result result, int rowNum) throws Exception {
                JSONArray results = new JSONArray();
                List<Cell> ceList = result.listCells();
                String res = "";
                if (ceList != null && ceList.size() > 0) {
                    byte[] f = new byte[]{" ".getBytes()[0]};

                    for (int i = 0; i < columnName.size(); ++i) {
                        res = null == Bytes.toString(null == result.getValue("basic".getBytes(), ((String) columnName.get(i)).getBytes()) ? f : result.getValue("basic".getBytes(), ((String) columnName.get(i)).getBytes())) ? "" : Bytes.toString(result.getValue("basic".getBytes(), ((String) columnName.get(i)).getBytes()));
                        if (StringUtils.isNotBlank(res)) {
                            results.add(JSONObject.parse(res));
                        }
                    }
                }
                return results;
            }
        });
        return result;
    }
}
